package com.young.sync.util;

import com.young.canal.serializer.EventRecord;
import com.young.springboot.kafka.KafkaClientConfiguration;
import com.young.springboot.kafka.KafkaClientProperties;
import com.young.sync.factory.ConsumeListenerFactory;
import com.young.sync.listener.MessageListener;
import com.young.sync.service.entity.TopicEntity;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.support.serializer.JsonDeserializer;
import org.springframework.stereotype.Component;

/**
 * @description：
 * @author: yangyahui01
 * @date: 12/5/24 3:27 PM
 */
@Component
@Slf4j
public class KafkaUtils {


    /**
     * 为 Topic 注册消费者
     * @param topic
     */
    public void registerMessageListener(TopicEntity topic) {

        log.info("注册topic监听:" + topic.toString());

        ConcurrentKafkaListenerContainerFactory<String, EventRecord> containerListenerByTopic = ConsumeListenerFactory.getContainerListenerByTopic(topic.getConsumeName());


        ConcurrentMessageListenerContainer<String, EventRecord> container = containerListenerByTopic.createContainer(topic.getTopic());

        MessageListener messageListener = new MessageListener(topic.getBizName());

        container.setBeanName(topic.getBizName());
        container.setupMessageListener(messageListener);


        container.start();

    }


    public ConcurrentKafkaListenerContainerFactory<String, EventRecord> registerConsumeFactory(KafkaClientProperties kafkaClientProperties) {

        KafkaClientConfiguration<String, EventRecord> clientConfiguration = new KafkaClientConfiguration<>(kafkaClientProperties);

        clientConfiguration.keyDeserializer(new StringDeserializer());
        clientConfiguration.valueDeserializer(new JsonDeserializer<>(EventRecord.class));

        ConcurrentKafkaListenerContainerFactory<String, EventRecord> containerFactory = clientConfiguration.kafkaListenerContainerFactory();

        containerFactory.setConsumerFactory(clientConfiguration.kafkaConsumerFactory()); // 这个可以注释掉了

        return containerFactory;
    }


}
